/**
 * Wihidum - Efficient Broker Hierarchy for Publish Subscribe Middleware
 *
 * License      :   Apache License 2.0
 * Home Page    :   http://code.google.com/p/wihidum/
 * Mailing List :   wihidum@googlegroups.com
 */

package lk.mrt.cse.wihidum.delivery;

import lk.mrt.cse.wihidum.storage.*;
import lk.mrt.cse.wihidum.util.*;
import lk.mrt.cse.wihidum.communicator.*;
import lk.mrt.cse.wihidum.processor.PublicationMessage;
import org.apache.axiom.om.*;
import org.apache.axiom.soap.*;
import org.apache.axis2.*;
import org.apache.axis2.wsdl.*;
import org.apache.axis2.context.*;
import org.apache.axis2.addressing.*;
import org.apache.axis2.client.*;
import org.apache.log4j.*;

import java.util.*;

/**
 * Thread to handle a single publication
 */
public class PublicationThread
        implements Runnable {

    private static final Logger logger = Logger.getLogger(PublicationThread.class);
    private PublicationMessage publicationMessage;
    private Router router;

    /**
     * Default constructor
     *
     * @param publicationMessage PublicationMessage
     */
    public PublicationThread(PublicationMessage publicationMessage) {
        this.publicationMessage = publicationMessage;
    }

    /**
     * Send the publication to interested subscribers
     */
    public void run() {
        if (WihidumConstants.currentTopology.equals(WihidumConstants.TopologyNames.clusterTopology)) {
            router = new ClusterRouter();
            router.propagatePublication(publicationMessage);
            this.deliverPublication();
        }
        if (WihidumConstants.currentTopology.equals(WihidumConstants.TopologyNames.treeTopology)) {
            router = new TreeRouter();
            // router.propagatePublication(publicationEnvelope, topic, WihidumConstants.Actions.Publish);
        }
    }

    private void deliverPublication() {
        try {
            ServiceClient serviceClient;
            serviceClient = new ServiceClient();

            //creating in out operating client
            OperationClient opClient = serviceClient.createClient(ServiceClient.ANON_OUT_IN_OP);
            //creating message context
            MessageContext outMsgCtx = new MessageContext();
            Options opts = outMsgCtx.getOptions();

            String SOAPVersion = opts.getSoapVersionURI();
            if (SOAPVersion == null)
                SOAPVersion = SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI;

            //Delivering publication to normal subscribers
            SubscriberManager subscriberDAO = new SubscriberDAO();
            ArrayList<Subscriber> subscriberList = subscriberDAO.getSubscribersFilteredByTopic((publicationMessage.getTopic()));

            for (Subscriber subscriber : subscriberList) {
                EndpointReference targetEPR = new EndpointReference(subscriber.getNotifyToAddress());
                opts.setTo(targetEPR);
                opts.setAction(WihidumConstants.Operations.STORE_PUBLICATION);

                SOAPEnvelope deliveryEnvelope = this.createPublicationDeliveryMessage(subscriber, SOAPVersion);

                outMsgCtx.setEnvelope(deliveryEnvelope);
                opClient.addMessageContext(outMsgCtx);
                opClient.execute(true);

                MessageContext inMsgtCtx = opClient.getMessageContext(WSDLConstants.MESSAGE_LABEL_IN_VALUE);

                SOAPEnvelope response = inMsgtCtx.getEnvelope();
                OMElement responseTextElement = response.getBody().getFirstElement();
                logger.info(responseTextElement.getText());
            }

            //Delivering publication to normal broker subscribers
            BrokerSubscriptionDAO brokerSubscriptionDAO = new BrokerSubscriptionDAO();
            ArrayList<BrokerSubscription> brokerSubscriptionList = brokerSubscriptionDAO.getBrokerSubscriptionsFilteredByTopic(publicationMessage.getTopic());

            for (BrokerSubscription brokerSubscription : brokerSubscriptionList) {
                logger.info("Preparing to send publication to broker subscriber : " + brokerSubscription.getFromBrokerAddress());
                if (!brokerSubscription.getFromBrokerAddress().equals(publicationMessage.getBrokerPublicationElementFrom())) {

                    //setting properties into option
                    opts.setTo(new EndpointReference(brokerSubscription.getFromBrokerAddress()));
                    opts.setAction(WihidumConstants.Operations.BROKER_PUBLISH);

                    NeighbourBroker neighbourBroker = new NeighbourBroker();
                    neighbourBroker.setAddress(brokerSubscription.getFromBrokerAddress());

                    //This is set to a constant because, Subscription process doesn't provide this value yet with the BrokerSubscription
                    neighbourBroker.setBroekrID("1.1.1.1");

                    // ClusterRouter clusterPropagator = new ClusterRouter();
                    SOAPEnvelope brokerPublicationMessage = router.createPublicationRequest(publicationMessage, neighbourBroker, SOAPVersion);

                    outMsgCtx.setEnvelope(brokerPublicationMessage);
                    opClient.addMessageContext(outMsgCtx);
                    logger.info("Sending publication to broker subscriber : " + brokerSubscription.getFromBrokerAddress());
                    opClient.execute(true);

                    MessageContext inMsgtCtx = opClient.getMessageContext(WSDLConstants.MESSAGE_LABEL_IN_VALUE);

                    SOAPEnvelope response = inMsgtCtx.getEnvelope();
                    OMElement responseTextElement = response.getBody().getFirstElement();
                } else {
                    logger.info("This publication has arrived from the same broker subscriber : " + brokerSubscription.getFromBrokerAddress() + " to which we are trying to diliver the broker publication. Hence, the delivery was cancelled.");
                }
            }
        }
        catch (NullPointerException ex) {
            logger.error("Error in processing store publication response");
            ex.printStackTrace();
        }
        catch (AxisFault ex) {
            logger.error("Error in sending publication", ex);
        }
    }

    /*
   /**
    * Create element to be sent to the listener
    *
    * @param SOAPURI SOAP version to be used in delivery message
    * @return Delivery element
    */
    private SOAPEnvelope createPublicationDeliveryMessage(Subscriber subscriber, String SOAPURI) {
        try {
            SOAPFactory soapFactory = null;
            if (SOAP11Constants.SOAP_ENVELOPE_NAMESPACE_URI.equals(SOAPURI))
                soapFactory = OMAbstractFactory.getSOAP11Factory();
            else if (SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI.equals(SOAPURI))
                soapFactory = OMAbstractFactory.getSOAP12Factory();
            else throw new Exception("Unknown SOAP version");

            SOAPEnvelope soapEnvelope = soapFactory.getDefaultEnvelope();

            //declare the used namespaces in the SOAP Envelope
            OMNamespace wseventingNS = soapFactory.createOMNamespace(WSEventingConstants.EVENTING_NAMESPACE, WSEventingConstants.EVENTING_PREFIX);
            soapEnvelope.declareNamespace(wseventingNS);
            OMNamespace wsaNS = soapFactory.createOMNamespace(WSEventingConstants.ADDRESSING_NAMESPACE, WSEventingConstants.ADDRESSING_PREFIX);
            soapEnvelope.declareNamespace(wsaNS);
            OMNamespace wihidumNS = soapFactory.createOMNamespace(WSEventingConstants.WIHIDUM_NAMESPACE, WSEventingConstants.WHIDUM_NAMESPACE_PREFIX);
            soapEnvelope.declareNamespace(wihidumNS);

            SOAPHeader soapHeader = soapEnvelope.getHeader();

            OMFactory factory = OMAbstractFactory.getOMFactory();


            OMElement publicationDeliveryHeader = factory.createOMElement("PublicationDeliveryHeader", wihidumNS);

            OMElement subscriptionIDElement = factory.createOMElement("SubscriptionID", wihidumNS);
            OMElement publisherAddressElement = factory.createOMElement("PublisherAddress", wihidumNS);
            OMElement topicElement = factory.createOMElement("Topic", wihidumNS);

            publisherAddressElement.setText(publicationMessage.getEventSourceURL());
            topicElement.setText(publicationMessage.getTopic());

            publicationDeliveryHeader.addChild(subscriptionIDElement);
            publicationDeliveryHeader.addChild(publisherAddressElement);
            publicationDeliveryHeader.addChild(topicElement);

            soapHeader.addChild(publicationDeliveryHeader);
            SOAPBody soapBody = soapEnvelope.getBody();
            soapBody.setText(publicationMessage.getBody());

            return soapEnvelope;
        }
        catch (Exception ex) {
            logger.error(ex);
            return null;
        }

    }
}
